package kmeans;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;

public class KMeansCluster {
    public static class KMeansClusterMapper extends
            Mapper<Object, Text, Text, IntWritable> {
        private final ArrayList<Cluster> kClusters = new ArrayList<Cluster>();

        @Override
        protected void setup(Context context) throws IOException {
            FileSystem fileSystem = FileSystem.get(context.getConfiguration());
            FileStatus[] fileStatuses = fileSystem.listStatus(new Path(context.getConfiguration().get("cluster Path")));
            for (FileStatus fileStatus : fileStatuses) {
                if (!fileStatus.isDirectory()) {
                    FSDataInputStream fsDataInputStream = fileSystem.open(fileStatus.getPath());
                    BufferedReader in = new BufferedReader(new InputStreamReader(fsDataInputStream));
                    String line = null;
                    while ((line = in.readLine()) != null) {
                        Cluster cluster = new Cluster(line);
                        kClusters.add(cluster);
                    }
                    in.close();
                    fsDataInputStream.close();
                }
            }
        }

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            Instance instance = new Instance(value.toString());
            int id = -1;
            double distance = Double.MAX_VALUE;
            for (Cluster cluster : kClusters) {
                double temdistance = cluster.ComputeDist(instance);
                if (distance > temdistance) {
                    id = cluster.getClusterId();
                    distance = temdistance;
                }
            }
            if (id == -1) {
                try {
                    throw new InterruptedException("id == -1");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    System.exit(-1);
                }
            } else {
                context.write(value, new IntWritable(id));
            }
        }
    }

    public static void run(String in, String out, String clusterPath)
            throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration(), "KMeansClusterJob");
        job.setJarByClass(KMeansCluster.class);
        job.setMapperClass(KMeansClusterMapper.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.getConfiguration().set("cluster Path", clusterPath);

        FileInputFormat.addInputPath(job, new Path(in));
        FileOutputFormat.setOutputPath(job, new Path(out));

        job.waitForCompletion(true);
    }
}